package com.atguigu.chapter11.hive;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.table.catalog.hive.HiveCatalog;

/**
 * Author: Pepsi
 * Date: 2023/9/4
 * Desc:
 */
public class Flink01_Hive {
    public static void main(String[] args) {
        // 解决权限异常的
        System.setProperty("HADOOP_USER_NAME","atguigu");

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);

        tEnv.executeSql("create table person(" +
                " id string, " +
                " ts bigint, " +
                " vc int " +
                ")with(" +
                " 'connector' = 'filesystem', " +
                " 'path'= 'input/sensor.json', " +
                " 'format' = 'json' " +
                ")");


        // 1. 创建一个hivecatalog
        HiveCatalog hiveCatalog = new HiveCatalog("hive", "studyTest", "input/");

        // 2. 向表环境注册hivecatalog
        tEnv.registerCatalog("hive",hiveCatalog);
//        tEnv.useCatalog("hive");
//        tEnv.useDatabase("studyTest");  //不想在sql里面加前缀就在这声明

        tEnv.useCatalog("default_catalog");
        tEnv.useDatabase("default_database");

        // 3. 使用hivecatalog去读取hive数据
        tEnv.sqlQuery("select " +
                " * " +
                "from person")
                .execute()
                .print();
    }
}
